package com.learning.hadoop.mapreduce.boost;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.text.NumberFormat;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class ThirdMapper extends Mapper<LongWritable, Text, Text, Text>{

    //存放微博总数, 将小数据缓存进内存, 预加载
        public static Map<String, Integer> cmap = null;
        //存放df
        public static Map<String, Integer> df = null;
        
        // 在初始化类时执行, 将数据预加载进map
        protected void setup(Context context)
                throws IOException, InterruptedException {
            
            System.out.println("*****");
            if (cmap == null || cmap.size() == 0 || df == null || df.size() == 0) {
                URI[] cacheFiles = context.getCacheFiles();
                if (cacheFiles != null) {
                    for (URI uri : cacheFiles) {
                        if (uri.getPath().endsWith("part-r-00003")) {
                            Path path = new Path(uri.getPath());
                            // 获取文件
                            Configuration configuration = context.getConfiguration();
                            FileSystem fs = FileSystem.get(configuration);
                            FSDataInputStream open = fs.open(path);
                            BufferedReader reader = new BufferedReader(new InputStreamReader(open));
                            
//                            BufferedReader reader = new BufferedReader(new FileReader(path.getName()));
                            String line = reader.readLine();
                            if (line.startsWith("count")) {
                                String[] split = line.split("\t");
                                cmap = new HashMap<>();
                                cmap.put(split[0], Integer.parseInt(split[1].trim()));
                            }
                            reader.close();
                        }else if (uri.getPath().endsWith("part-r-00000")) {
                            df = new HashMap<>();
                            Path path = new Path(uri.getPath());
                            
                            // 获取文件
                            Configuration configuration = context.getConfiguration();
                            FileSystem fs = FileSystem.get(configuration);
                            FSDataInputStream open = fs.open(path);
                            BufferedReader reader = new BufferedReader(new InputStreamReader(open));
//                            BufferedReader reader = new BufferedReader(new FileReader(path.getName()));
                            
                            String line = null;
                            while ((line = reader.readLine()) != null) {
                                String[] ls = line.split("\t");
                                df.put(ls[0], Integer.parseInt(ls[1].trim()));
                            }
                            reader.close();
                        }
                    }
                }
            }
        }
        
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            // 获取分片
            FileSplit inputSplit = (FileSplit) context.getInputSplit();
            
            if (!inputSplit.getPath().getName().contains("part-r-00003")) {
                String[] values = value.toString().trim().split("\t");
                
                if (values.length >= 2) {
                    
                    int tf = Integer.parseInt(values[1].trim());
                    
                    String[] ss = values[0].split("_");
                    if (ss.length >= 2) {
                        String word = ss[0];
                        String id = ss[1];
                        
                        // 公式
                        Double s = tf * Math.log(cmap.get("count")) / df.get(word);
                        NumberFormat format = NumberFormat.getInstance();
                        // 取小数点后5位
                        format.setMaximumFractionDigits(5);
                        
                        context.write(new Text(id), new Text(word + ": " + format.format(s)));
                    }else {
                        System.out.println(value.toString() + "------");
                    }
                }
            }
        }
}